package com.ibm.cps.spark.streaming;

import backtype.storm.tuple.Values;
import com.fasterxml.jackson.databind.node.ObjectNode;
import com.google.common.base.Throwables;
import com.ibm.cps.message.MessageFactory;
import com.ibm.cps.message.MetadataTopologySort;
import com.ibm.cps.newmessage.*;
import com.ibm.cps.spark.streaming.adapter.SparkEmbeddedCollector;
import com.ibm.cps.storm.DataSourceBolt;
import com.ibm.cps.storm.DispatchBolt;
import com.ibm.cps.storm.ProcessorBolt;
import com.ibm.interfaces.IPersistency;
import com.ibm.mongo.MongoPersistency;
import com.ibm.util.TopologyStreamIds;
import com.ibm.util.exception.CPSException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import com.ibm.util.JSONUtility;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Iterator;
import java.util.List;

/**
 * Created by telekinesis on 9/10/15.
 */
public class EmbeddedTopologyCreator {
    private static IPersistency persistency;
    private EmbeddedTopologyCreator() throws CPSException {
    }

    private static final Logger logger = LoggerFactory.getLogger(EmbeddedTopologyCreator.class);

    public static SparkEmbeddedCollector create(String userId, int taskId) throws CPSException{
        SparkEmbeddedCollector collector = new SparkEmbeddedCollector();

        DispatchBolt dispatchBolt = new DispatchBolt();
        dispatchBolt.prepare(null, null, collector);
        ProcessorBolt processorBolt = new ProcessorBolt();
        processorBolt.manuallyPrepare(taskId, collector);
        DataSourceBolt datasourceBolt = new DataSourceBolt();
        datasourceBolt.prepare(null, null, collector);

        collector.addBolt(TopologyStreamIds.PROCESSOR_SPOUT_STREAM, dispatchBolt);
        collector.addBolt(TopologyStreamIds.PROCESSOR_DELETE_SPOUT_STREAM, dispatchBolt);
        collector.addBolt(TopologyStreamIds.PARSED_DATA_STREAM, dispatchBolt);
        collector.addBolt(TopologyStreamIds.STREAM_TO_DISPATCH, dispatchBolt);

        collector.addBolt(TopologyStreamIds.METADATA_STREAM, processorBolt);
        collector.addBolt(TopologyStreamIds.METADATA_DELETE_STREAM, processorBolt);
        collector.addBolt(TopologyStreamIds.DATAMESSAGE_STREAM, processorBolt);
        collector.addBolt(TopologyStreamIds.SPARK_BATCH_ENDING, processorBolt);

        collector.addBolt(TopologyStreamIds.DATASOURCE_SPOUT_STREAM, datasourceBolt);
        collector.addBolt(TopologyStreamIds.DATASOURCE_DELETE_SPOUT_STREAM, datasourceBolt);
        collector.addBolt(TopologyStreamIds.RAW_DATA_STREAM, datasourceBolt);

        try {
            List<SerializableDataSourceMetadata> existingDatasource = loadDatasource(userId);
            for(SerializableDataSourceMetadata source : existingDatasource){
                collector.emit(TopologyStreamIds.DATASOURCE_SPOUT_STREAM, new Values(source));
            }
            List<AbstractMetadata> existingProcessors = loadProcessors(userId);
            for(AbstractMetadata processor : existingProcessors){
                collector.emit(TopologyStreamIds.PROCESSOR_SPOUT_STREAM, new Values(processor));
            }
        }catch(CPSException ex){
            logger.error(Throwables.getStackTraceAsString(ex));
        }

        return collector;
    }
    public static SparkEmbeddedCollector create(int taskId) throws CPSException{
        SparkEmbeddedCollector collector = new SparkEmbeddedCollector();
        persistency = MongoPersistency.getInstace();

        DispatchBolt dispatchBolt = new DispatchBolt();
        dispatchBolt.prepare(null, null, collector);
        ProcessorBolt processorBolt = new ProcessorBolt();
        processorBolt.manuallyPrepare(taskId, collector);
        DataSourceBolt datasourceBolt = new DataSourceBolt();
        datasourceBolt.prepare(null, null, collector);

        collector.addBolt(TopologyStreamIds.PROCESSOR_SPOUT_STREAM, dispatchBolt);
        collector.addBolt(TopologyStreamIds.PROCESSOR_DELETE_SPOUT_STREAM, dispatchBolt);
        collector.addBolt(TopologyStreamIds.PARSED_DATA_STREAM, dispatchBolt);
        collector.addBolt(TopologyStreamIds.STREAM_TO_DISPATCH, dispatchBolt);

        collector.addBolt(TopologyStreamIds.METADATA_STREAM, processorBolt);
        collector.addBolt(TopologyStreamIds.METADATA_DELETE_STREAM, processorBolt);
        collector.addBolt(TopologyStreamIds.DATAMESSAGE_STREAM, processorBolt);
        collector.addBolt(TopologyStreamIds.SPARK_BATCH_ENDING, processorBolt);

        collector.addBolt(TopologyStreamIds.DATASOURCE_SPOUT_STREAM, datasourceBolt);
        collector.addBolt(TopologyStreamIds.DATASOURCE_DELETE_SPOUT_STREAM, datasourceBolt);
        collector.addBolt(TopologyStreamIds.RAW_DATA_STREAM, datasourceBolt);

        try {
            Collection<String> tenants = persistency.getCredential(JSONUtility.newObjectNode());
            Iterator<String> it = tenants.iterator();
            String userId = new String();
            while(it.hasNext()) {
                String credential = it.next();
                String tenant_type = credential.split("\"tenant_type\":\"")[1].split("\",")[0];
                if(tenant_type.equals("normalTenant")) {
                    userId = credential.split("\"tenantid\":\"")[1].split("\"}")[0];
                    List<SerializableDataSourceMetadata> existingDatasource = loadDatasource(userId);
                    for(SerializableDataSourceMetadata source : existingDatasource){
                        collector.emit(TopologyStreamIds.DATASOURCE_SPOUT_STREAM, new Values(source));
                    }
                    List<AbstractMetadata> existingProcessors = loadProcessors(userId);
                    for(AbstractMetadata processor : existingProcessors){
                        collector.emit(TopologyStreamIds.PROCESSOR_SPOUT_STREAM, new Values(processor));
                    }
                }

            }


        }catch(CPSException ex){
            logger.error(Throwables.getStackTraceAsString(ex));
        }

        return collector;
    }
    private static List<AbstractMetadata> loadProcessors(String tenantid) throws CPSException{
        List<AbstractMetadata> result = new ArrayList<>();
        Collection<String> messages = MessageFactory.loadExistedMessage(tenantid);
        List<ObjectNode> outputs = MetadataTopologySort.getSortedMetadatas(messages);
        if (outputs != null) {
            for (ObjectNode metadata : outputs) {
                logger.info("Load processor " + metadata.toString());
                AbstractMetadata message = MetadataFactory.parseJsonMetadata(metadata.toString());
                result.add(message);
            }
        }
        return result;
    }

    private static List<SerializableDataSourceMetadata> loadDatasource(String tenantid) throws CPSException{
        List<SerializableDataSourceMetadata> result = new ArrayList<>();
        Collection<String> existedDs = DataSourceMetadataFactory.loadExistedDataSource(tenantid);
        if (existedDs != null) {
            for (String datasource : existedDs) {
                SerializableDataSourceMetadata dsMetadata = new SerializableDataSourceMetadata(
                        datasource);
                result.add(dsMetadata);
            }
        }
        return result;
    }
}
